package com.yanfan.mqtt.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.yanfan.base.service.ISessionStore;
import com.yanfan.base.session.Session;
import com.yanfan.common.constant.YanfanConstant;
import com.yanfan.common.core.mq.DeviceReplyBo;
import com.yanfan.common.core.mq.DeviceReport;
import com.yanfan.common.core.mq.DeviceReportBo;
import com.yanfan.common.core.mq.message.DeviceData;
import com.yanfan.common.core.mq.ota.OtaReplyMessage;
import com.yanfan.common.core.mq.ota.OtaUpgradeBo;
import com.yanfan.common.core.redis.RedisCache;
import com.yanfan.common.core.redis.RedisKeyBuilder;
import com.yanfan.common.core.thingsModel.ThingsModelSimpleItem;
import com.yanfan.common.enums.*;
import com.yanfan.common.exception.ServiceException;
import com.yanfan.common.utils.DateUtils;
import com.yanfan.common.utils.gateway.mq.TopicsUtils;
import com.yanfan.framework.manager.AsyncManager;
import com.yanfan.iot.domain.Device;
import com.yanfan.iot.domain.FunctionLog;
import com.yanfan.iot.enums.DeviceType;
import com.yanfan.iot.model.DeviceStatusVO;
import com.yanfan.iot.model.ThingsModels.ValueItem;
import com.yanfan.iot.service.IDeviceService;
import com.yanfan.iot.service.IFirmwareTaskDetailService;
import com.yanfan.iot.service.IFunctionLogService;
import com.yanfan.iot.service.IProductService;
import com.yanfan.mq.model.ReportDataBo;
import com.yanfan.mq.service.IDataHandler;
import com.yanfan.mq.service.IDeviceReportMessageService;
import com.yanfan.mq.service.IMqttMessageSubscribeAware;
import com.yanfan.mqtt.manager.MqttRemoteManager;
import com.yanfan.mqtt.model.PushMessageBo;
import com.yanfan.protocol.base.protocol.IProtocol;
import com.yanfan.protocol.service.IProtocolManagerService;
import io.netty.buffer.Unpooled;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.*;

/**
 * 处理类 处理设备主动上报和设备回调信息
 *
 * @author bill
 */
@Service
@Slf4j
public class DeviceReportMessageServiceImpl implements IDeviceReportMessageService {

    @Autowired
    private IDeviceService deviceService;
    @Autowired
    private IProductService productService;
    @Autowired
    private IProtocolManagerService protocolManagerService;
    @Autowired
    private IFirmwareTaskDetailService firmwareTaskDetailService;
    @Resource
    private TopicsUtils topicsUtils;
    @Resource
    private IDataHandler dataHandler;
    @Resource
    private MqttRemoteManager remoteManager;
    @Resource
    private RedisCache redisCache;
    @Resource
    private IFunctionLogService functionLogService;
    @Resource
    private MqttRemoteManager mqttRemoteManager;
    @Resource
    private ISessionStore sessionStore;
    @Autowired
    private IMqttMessageSubscribeAware messageSubscribeAware;
    /**
     * 处理设备主动上报数据  ,这是读取刚刚那个队列的函数
     */
    @Override
    public void parseReportMsg(DeviceReportBo bo) {
        String serialNumber = bo.getSerialNumber();
        switch (bo.getServerType()) {
            case MQTT:
                log.debug("=>MQ*收到设备主题[{}],消息:[{}]", bo.getTopicName(), bo.getData());
                //构建消息
                DeviceStatusVO deviceStatusVO = deviceService.selectDeviceStatusAndTransportStatus(serialNumber);
                Optional.ofNullable(deviceStatusVO).orElseThrow(() -> new ServiceException("设备不存在"));
                Long productId = deviceStatusVO.getProductId();
                /*获取协议处理器*/
                IProtocol protocol = selectedProtocol(productId);
                DeviceData data = DeviceData
                        .builder()
                        .serialNumber(serialNumber)
                        .topicName(bo.getTopicName())
                        .productId(productId)
                        .data(bo.getData())
                        .buf(Unpooled.wrappedBuffer(bo.getData()))
                        .build();
                /*根据协议解析后的数据*/
                DeviceReport reportMessage = protocol.decode(data, serialNumber);
                //设备回复，更新指令下发记录
                if (reportMessage.getIsReply()) {
                    handlerDeviceReply(reportMessage);
                }
                serialNumber = reportMessage.getSerialNumber() == null ? serialNumber : reportMessage.getSerialNumber();
                productId = reportMessage.getProductId() == null ? productId : reportMessage.getProductId();
                reportMessage.setSerialNumber(serialNumber);
                reportMessage.setProductId(productId);
                reportMessage.setPlatformDate(bo.getPlatformDate());
                reportMessage.setServerType(bo.getServerType());
                //同步设备状态
                this.synchDeviceStatus(serialNumber);
                //处理网关设备上报数据
                processNoSub(reportMessage, bo.getTopicName());
                break;
            case TCP:
                log.debug("*MQ收到TCP推送消息[{}]", JSON.toJSON(bo.getThingsModelSimpleItem()));
                DeviceStatusVO deviceStatusVO1 = deviceService.selectDeviceStatusAndTransportStatus(serialNumber);
                Optional.ofNullable(deviceStatusVO1).orElseThrow(() -> new ServiceException("设备不存在"));
                //同步设备状态
                this.synchDeviceStatus(serialNumber);
                DeviceReport deviceReport = new DeviceReport();
                BeanUtils.copyProperties(bo, deviceReport);
                deviceReport.setProductId(deviceStatusVO1.getProductId());
                deviceReport.setThingsModelSimpleItem(bo.getThingsModelSimpleItem());
                deviceReport.setSlaveId(bo.getSlaveId());
                deviceReport.setSerialNumber(deviceStatusVO1.getSerialNumber());
                //设备回复数据处理
                if (bo.getIsReply()) {
                    handlerDeviceReply(deviceReport);
                }
                processNoSub(deviceReport, null);
                break;
        }
    }

    /**
     * 处理设备回调数据
     */
    @Override
    public void parseReplyMsg(DeviceReportBo bo) {
        log.debug("=>MQ*收到设备回调消息,[{}]", bo);
        buildReport(bo);
        //获取解析协议
        IProtocol protocol = selectedProtocol(bo.getProductId());
        DeviceData deviceSource = DeviceData
                .builder()
                .serialNumber(bo.getSerialNumber())
                .topicName(bo.getTopicName())
                .data(bo.getData())
                .build();
        //协议解析后的数据
        DeviceReport message = protocol.decode(deviceSource, null);
        //处理网关设备回复数据
        processNoSub(message, bo.getTopicName());

    }

    /**
     * 处理设备OTA升级
     *
     * @param bo
     */
    @Override
    public void parseOTAUpdateReply(DeviceReportBo bo) {
        //构建消息
        buildReport(bo);
        //获取编码协议器
        IProtocol protocol = selectedProtocol(bo.getProductId());
        DeviceData deviceSource = DeviceData
                .builder()
                .serialNumber(bo.getSerialNumber())
                .topicName(bo.getTopicName())
                .data(bo.getData())
                .build();
        DeviceReport otaMessage = protocol.decode(deviceSource, null);
        otaMessage.setSerialNumber(bo.getSerialNumber());
        otaMessage.setProductId(bo.getProductId());
        otaMessage.setMessageId(bo.getMessageId());
        //处理OTA升级回复消息
        OtaUpgradeBo upgradeBo = OtaUpgradeBo.builder().messageId(otaMessage.getMessageId()).build();
        JSONObject result = JSONObject.parseObject(otaMessage.getReplyMessage());
        /*
           根据各自设备自定义的OTA升级回复来处理数据
           这里约定设备OTA升级，返回消息结构如下
           json结构: "data":{"code":1,"msg":"true"}
         */
        JSONObject data = (JSONObject) result.get("data");
        OtaReplyMessage otaReplyMessage = JSONObject.toJavaObject(data, OtaReplyMessage.class);
        otaReplyMessage.setMessageId(otaMessage.getMessageId());
        OTAUpgrade upgrade = OTAUpgrade.parse(otaReplyMessage.getCode());
        //更新数据库 OTA升级状态
        String version = firmwareTaskDetailService.update(upgradeBo, upgrade);
        //如果OTA升级成功，更新设备的固件版本号
        if (upgrade.equals(OTAUpgrade.SUCCESS)) {
            Device device = new Device();
            device.setSerialNumber(bo.getSerialNumber());
            device.setFirmwareVersion(new BigDecimal(version));
            deviceService.updateDeviceFirmwareVersion(device);
        }

    }

    /**
     * 构建消息
     *
     * @param bo
     */
    @Override
    public DeviceStatusVO buildReport(DeviceReportBo bo) {
        DeviceStatusVO deviceStatusVO = deviceService.selectDeviceStatusAndTransportStatus(bo.getSerialNumber());
        Optional.ofNullable(deviceStatusVO).orElseThrow(() -> new ServiceException("设备不存在"));
        //产品id
        bo.setProductId(deviceStatusVO.getProductId());
        return deviceStatusVO;
    }

    /**
     * 根据产品id获取协议处理器
     */
    @Override
    public IProtocol selectedProtocol(Long productId) {
        //查询产品获取协议编号
        String code = productService.getProtocolByProductId(productId);
        return protocolManagerService.getProtocolByProtocolCode(code);
    }

    /**
     * 处理网关设备
     *
     * @param message
     * @param topicName
     */
    /**
     * 处理网关设备
     *
     * @param message
     * @param topicName
     */
    @Override
    public void processNoSub(DeviceReport message, String topicName) {
        if (message.getServerType().equals(ServerType.MQTT)) {
            //处理topic以prop结尾上报的数据 (属性)
            if (!topicName.endsWith(TopicType.PROPERTY_POST.getTopicSuffix())) {
                return;
            }
        }

        ReportDataBo report = new ReportDataBo();
        report.setSerialNumber(message.getSerialNumber());
        report.setProductId(message.getProductId());
        report.setDataList(message.getThingsModelSimpleItem());
        report.setType(1);
        report.setUserId(message.getUserId());
        report.setUserName(message.getUserName());
        report.setDeviceName(message.getDeviceName());
        report.setSources(message.getSources());
        //属性上报执行规则引擎
        report.setRuleEngine(true);
        dataHandler.reportData(report);

        // 异步执行
        AsyncManager.me().execute(new TimerTask() {
            @Override
            public void run() {
                messageSubscribeAware.publishMessage(message, topicName);
            }
        });
    }


    /**
     * 处理设备回调信息，此处按照topic区分 prop上报和设备回调reply，
     * 如果模组可订阅的topic有限，不能区分prop上报和reply，自行根据上报数据来区分
     *
     * @param message
     */
    public void handlerDeviceReply(DeviceReport message) {
        String messageId = "";
        String sources = message.getSources();
        String serialNumber = message.getSerialNumber();
        Long productId = message.getProductId();
        String cacheKey = RedisKeyBuilder.buildDownMessageIdCacheKey(serialNumber);
        Set<String> functionList = redisCache.zRange(cacheKey, 0, -1);
        //从redis中获取messageId 流水号，获取下发记录
        for (String fun : functionList) {
            String[] split = fun.split(":");
            if (split[0].equals(sources)) {
                messageId = split[1];
            }
            redisCache.zRem(cacheKey, fun);
        }
        FunctionLog functionLog = new FunctionLog();
        //发布设备回复消息，通知前端
        PushMessageBo pushMessageBo = new PushMessageBo();
        JSONObject jsonObject = new JSONObject();
        //处理mqtt设备的回复数据
        switch (message.getServerType()) {
            case MQTT:
                pushMessageBo.setTopic(topicsUtils.buildTopic(productId, serialNumber, TopicType.SERVICE_INVOKE_REPLY));
                switch (message.getProtocolCode()) {
                    //银尔达开关设备回复处理
                    case YanfanConstant.PROTOCOL.YinErDa:
                        jsonObject.put("message", message.getReplyMessage());
                        pushMessageBo.setMessage(jsonObject.toJSONString());
                        remoteManager.pushCommon(pushMessageBo);
                        //更新设备值
                        String messageIdCacheKey = RedisKeyBuilder.buildDownMessageIdCacheKey(serialNumber);
                        DeviceReplyBo replyBo = redisCache.getCacheObject(messageIdCacheKey);
                        String cacheKey1 = RedisKeyBuilder.buildTSLVCacheKey(productId, serialNumber);
                        ValueItem valueItem = redisCache.getCacheMapValue(cacheKey1, replyBo.getId());
                        valueItem.setShadow(replyBo.getValue());
                        valueItem.setValue(replyBo.getValue());
                        valueItem.setTs(DateUtils.getNowDate());
                        redisCache.setCacheMapValue(cacheKey1, replyBo.getId(), com.alibaba.fastjson2.JSONObject.toJSONString(valueItem));
                        break;
                    //modbus-rtu设备回复数据处理
                    case YanfanConstant.PROTOCOL.ModbusRtu:
                        String mqttMsgCacheKey = RedisKeyBuilder.buildDownMessageIdCacheKey(message.getSerialNumber());
                        DeviceReplyBo mqttReply = redisCache.getCacheObject(mqttMsgCacheKey);
                        if (!Objects.isNull(mqttReply) && mqttReply.getId().equals(message.getAddress() + "")) {
                            functionLog.setResultCode(FunctionReplyStatus.SUCCESS.getCode());
                            functionLog.setResultMsg(FunctionReplyStatus.SUCCESS.getMessage());
                            functionLog.setReplyTime(DateUtils.getNowDate());
                            functionLog.setMessageId(mqttReply.getMessageId());
                            functionLogService.updateByMessageId(functionLog);
                        } else {
                            break;
                        }
                        List<ThingsModelSimpleItem> result = new ArrayList<>();
                        ThingsModelSimpleItem item = new ThingsModelSimpleItem();
                        item.setId(message.getAddress() + "");
                        item.setSlaveId(message.getSlaveId());
                        item.setTs(DateUtils.getNowDate());
                        item.setValue(message.getReplyMessage());
                        item.setRemark(DateUtils.getTime());
                        result.add(item);
                        pushMessageBo.setMessage(JSON.toJSONString(result));
                        pushMessageBo.setTopic(topicsUtils.buildTopic(message.getProductId(), message.getSerialNumber(), TopicType.SERVICE_INVOKE_REPLY));
                        remoteManager.pushCommon(pushMessageBo);
                        String modbusCacheKey = RedisKeyBuilder.buildTSLVCacheKey(productId, serialNumber);
                        String hkey = message.getAddress() + "#" + message.getSlaveId();
                        String cacheMapValue = redisCache.getCacheMapValue(modbusCacheKey, hkey);
                        ValueItem modbusItem = com.alibaba.fastjson2.JSON.parseObject(cacheMapValue, ValueItem.class);
                        if (Objects.isNull(modbusItem)) {
                            modbusItem = new ValueItem();
                            modbusItem.setId(message.getAddress() + "");
                            modbusItem.setSlaveId(message.getSlaveId());
                            modbusItem.setRegArr(null);
                        }
                        modbusItem.setShadow(message.getReplyMessage());
                        modbusItem.setValue(message.getReplyMessage());
                        modbusItem.setTs(DateUtils.getNowDate());
                        redisCache.setCacheMapValue(modbusCacheKey, hkey, com.alibaba.fastjson2.JSONObject.toJSONString(modbusItem));
                        break;
                    case YanfanConstant.PROTOCOL.ModbusToJsonHP:
                    case YanfanConstant.PROTOCOL.ModbusRtuPak:
                        functionLog.setResultCode(FunctionReplyStatus.SUCCESS.getCode());
                        functionLog.setResultMsg(FunctionReplyStatus.SUCCESS.getMessage());
                        functionLog.setReplyTime(DateUtils.getNowDate());
                        functionLog.setMessageId(message.getMessageId() == null ? messageId : message.getMessageId());
                        functionLogService.updateByMessageId(functionLog);
                        //推送到前端更新指令下发日志记录
                        jsonObject.put("update", 1);
                        pushMessageBo.setTopic(topicsUtils.buildTopic(productId, serialNumber, TopicType.SERVICE_INVOKE_REPLY));
                        pushMessageBo.setMessage(jsonObject.toJSONString());
                        remoteManager.pushCommon(pushMessageBo);
                }
                break;
            case TCP:
                switch (message.getProtocolCode()) {
                    //modbus-rtu设备回复数据处理
                    case YanfanConstant.PROTOCOL.ModbusRtu:
                        //更新指令下发日志
                        //更新设备值
                        String messageIdCacheKey = RedisKeyBuilder.buildDownMessageIdCacheKey(message.getSerialNumber());
                        DeviceReplyBo replyBo = redisCache.getCacheObject(messageIdCacheKey);
                        if (replyBo.getId().equals(message.getAddress() + "")) {
                            functionLog.setResultCode(FunctionReplyStatus.SUCCESS.getCode());
                            functionLog.setResultMsg(FunctionReplyStatus.SUCCESS.getMessage());
                            functionLog.setReplyTime(DateUtils.getNowDate());
                            functionLog.setMessageId(replyBo.getMessageId());
                            functionLogService.updateByMessageId(functionLog);
                        }
                        List<ThingsModelSimpleItem> result = new ArrayList<>();
                        ThingsModelSimpleItem item = new ThingsModelSimpleItem();
                        item.setId(message.getAddress() + "");
                        item.setSlaveId(message.getSlaveId());
                        item.setTs(DateUtils.getNowDate());
                        item.setValue(message.getReplyMessage());
                        item.setRemark(DateUtils.getTime());
                        result.add(item);
                        pushMessageBo.setMessage(JSON.toJSONString(result));
                        pushMessageBo.setTopic(topicsUtils.buildTopic(message.getProductId(), message.getSerialNumber(), TopicType.WS_SERVICE_INVOKE));
                        remoteManager.pushCommon(pushMessageBo);
                        String modbusCacheKey = RedisKeyBuilder.buildTSLVCacheKey(message.getProductId(), message.getSerialNumber());
                        String hkey = message.getAddress() + "#" + message.getSlaveId();
                        String cacheMapValue = redisCache.getCacheMapValue(modbusCacheKey, hkey);
                        ValueItem modbusItem = com.alibaba.fastjson2.JSON.parseObject(cacheMapValue, ValueItem.class);
                        if (Objects.isNull(modbusItem)) {
                            modbusItem = new ValueItem();
                            modbusItem.setId(message.getAddress() + "");
                            modbusItem.setSlaveId(message.getSlaveId());
                            modbusItem.setRegArr(null);
                        }
                        modbusItem.setShadow(message.getReplyMessage());
                        modbusItem.setValue(message.getReplyMessage());
                        modbusItem.setTs(DateUtils.getNowDate());
                        redisCache.setCacheMapValue(modbusCacheKey, hkey, com.alibaba.fastjson2.JSONObject.toJSONString(modbusItem));
                        break;

                }
                break;

        }
    }

    /**
     * 解析OTA升级回复消息,更新升级状态
     */
    private void otaUpgrade(DeviceReport message, String topicName) {


    }

    /**
     * 同步设备状态
     *
     * @param
     */
    private void synchDeviceStatus(String serialNumber) {
        //如果有数据上报，但是数据库设备状态为离线，则进行同步
        DeviceStatusVO deviceStatusVO = deviceService.selectDeviceStatusAndTransportStatus(serialNumber);
        if (deviceStatusVO.getStatus() == DeviceStatus.OFFLINE.getType() || deviceStatusVO.getStatus() == DeviceStatus.UNACTIVATED.getType()) {
            Device updateBo = new Device();
            updateBo.setStatus(3);
            updateBo.setSerialNumber(deviceStatusVO.getSerialNumber());
            updateBo.setUpdateTime(DateUtils.getNowDate());
            deviceService.updateDeviceStatus(updateBo);
            mqttRemoteManager.pushDeviceStatus(deviceStatusVO.getProductId(), deviceStatusVO.getSerialNumber(), DeviceStatus.ONLINE);

            //如果是子设备，维护子设备的状态到session
            if (deviceStatusVO.getDeviceType() == DeviceType.SUB_GATEWAY.getCode()) {
                Session session = new Session();
                session.setServerType(ServerType.MQTT);
                session.setClientId(deviceStatusVO.getSerialNumber());
                session.setLastAccessTime(DateUtils.getTimestamp());
                session.setConnected(true);
                sessionStore.storeSession(deviceStatusVO.getSerialNumber(), session);
            }
        }
    }
}
